package com.ithuameng.cosy.mq.listener;

import com.ithuameng.cosy.constants.Constants;
import com.ithuameng.cosy.group.SessionGroup;
import com.ithuameng.cosy.model.Message;
import com.ithuameng.cosy.util.JSONUtils;
import com.ithuameng.cosy.util.RedisCache;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Component
public class PushMessageListener {

    @Resource
    private RedisCache redisCache;

    @Resource
    private SessionGroup sessionGroup;

    @RabbitListener(queues = Constants.PUSH_MESSAGE_INNER_QUEUE)
    public void pushMessageListener(com.rabbitmq.client.Channel rabbitChannel, org.springframework.amqp.core.Message message) throws IOException {

        // 判断此消息是否已经消费过
        String msgId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");

        long tag = message.getMessageProperties().getDeliveryTag();

        if (redisCache.getCacheMap(Constants.MESSAGE_CONSUME_REDIS_RECORD).containsKey(msgId)) {

            log.info("[" + msgId + "] message has been consumed from PushMessageListener.");

            rabbitChannel.basicAck(tag, false);

            return;
        }

        try {
            Message msg = JSONUtils.fromJson(message.getBody(), Message.class);

            String uid = msg.getReceiver();

            sessionGroup.write(uid, msg);
        } catch (Exception e) {

            // 消息处理失败 true阻止消息重新回到队列中再次消费
            log.info("[" + msgId + "] message processing failure from PushMessageListener : " + e.getMessage());

            rabbitChannel.basicNack(tag, false, true);
        }

        // 标记消息被处理过
        Map<String, String> data = new HashMap<>(1);

        data.put(msgId, Constants.BIND_MESSAGE_ROUTING_KEY);

        redisCache.setCacheMap(Constants.MESSAGE_CONSUME_REDIS_RECORD, data);

        rabbitChannel.basicAck(tag, false);
    }
}
